package com.pig4cloud.pigx.contract.websocket.subscribe;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.pig4cloud.pigx.contract.enums.ContractOrderEnum;
import com.pig4cloud.pigx.contract.mongo.entity.ContractKLineTradeCacheMongo;
import com.pig4cloud.pigx.contract.mongo.entity.ContractSymbolMongo;
import com.pig4cloud.pigx.contract.utils.ContractUtils;
import com.pig4cloud.pigx.contract.vo.AbstractTopicVO;
import com.pig4cloud.pigx.contract.vo.Contract24hTradeWebsocketVO;
import com.pig4cloud.pigx.contract.vo.ContractTradeWebsocketVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 */
@Slf4j
@Component
public class MergeKLine extends DataListAbstractTopic {

	protected String TOPICNAME = "mergeKLine";

	private final MongoTemplate mongo;
	private final ContractUtils utils;
	@Autowired
	private RedisTemplate redis;

	public MergeKLine(MongoTemplate mongo, ContractUtils utils) {
		this.mongo = mongo;
		this.utils = utils;
		init();
	}

	private void init() {
		wsPush();
	}

	private void wsPush() {
		ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
		service.scheduleAtFixedRate(() -> {
			try {
				// ----------该推送方案 仅限于推送信息与用户信息无关联的场景----------
				// ----------实现方式为 将所有session根据查询类型进行划分,大幅度减少查询消耗,可承受大面积用户同时访问----------
				// ----------在服务器允许的情况下 可以将每种数据类型都new一个线程去执行,可大幅度提升并发访问的推送效率----------
				// 发送消息容器
				Map<String, List<WebSocketSession>> map = new HashMap<>();
				// 用户原生请求容器
				Map<String, List<String>> paramsMap = new HashMap<>();
				// 统计所有订阅用户的订阅类型
				for (Map.Entry<WebSocketSession, AbstractTopicVO> entry : subList.entrySet()) {
					if (!entry.getKey().isOpen()) {
						subList.remove(entry.getKey());
						log.info("ws 移除已关闭的session");
						continue;
					}

					// 请求参数解析
					AbstractTopicVO param = entry.getValue();
					List<String> dataList = param.getDataList();
					if (dataList.size() < 3) {
						continue;
					}

					// 查询币对为key
					String key = dataList.get(2);
					// 如果已有该类型,则在该类型中添加发送用户
					if (map.containsKey(key)) {
						map.get(key).add(entry.getKey());
					} else {
						List<WebSocketSession> sessionList = new ArrayList<>();
						sessionList.add(entry.getKey());
						map.put(key, sessionList);
						// 保存用户原生请求
						paramsMap.put(key, dataList);
					}
				}
				// 遍历所有需要发送的数据类型,将各类型数据分发给订阅的用户
				map.forEach((key, value) -> {
					// 参数组装
					JSONObject object = new JSONObject();
					// 最新成交
					JSONArray newestTrade = getNewestTrade(key);
					// 24h行情
					Contract24hTradeWebsocketVO riseAndFall = getRiseAndFall(key);
					// 拼接
					object.put("newestTrade", newestTrade);
					object.put("riseAndFall", riseAndFall);
					TextMessage message = getTextMessage(paramsMap.get(key), object);
					// 将数据发送给所有订阅该K线的用户
					value.forEach(session -> {
						try {
							session.sendMessage(message);
						} catch (IOException e) {
							e.printStackTrace();
							log.info("push merge kline err:" + e.getMessage());
						}
					});
				});
			} catch (Exception e) {
				e.printStackTrace();
				log.info("push merge kline err:" + e.getMessage());
			}

		}, 0, 200, TimeUnit.MILLISECONDS);
	}

	/**
	 * 获取24h成交
	 *
	 * @param symbolName 币对名称
	 * @return
	 */
	private Contract24hTradeWebsocketVO getRiseAndFall(String symbolName) {
		Query query = new Query();
		query.addCriteria(Criteria.where("name").is(symbolName));
		ContractSymbolMongo one = mongo.findOne(query, ContractSymbolMongo.class);
		Contract24hTradeWebsocketVO vo = new Contract24hTradeWebsocketVO();
		assert one != null;
		BeanUtils.copyProperties(one, vo);
		utils.setScaleBySymbol(vo, symbolName);
		return vo;
	}

	/**
	 * 获取最新成交记录
	 *
	 * @param symbolName 币对名称
	 * @return
	 */
	private JSONArray getNewestTrade(String symbolName) {
		// mongo集合名称
		String collectionName = symbolName + "_newest_trade";
		Query query = new Query();
		query.with(Sort.by(Sort.Order.desc("createTime")));
		// 读mongo缓存，读不到再读数据库
		List<ContractKLineTradeCacheMongo> vos = mongo.find(query, ContractKLineTradeCacheMongo.class, collectionName);
		List<ContractTradeWebsocketVO> websocketVO = new ArrayList<>();
		vos.forEach(e -> {
			try {
				Date time = Date.from(e.getCreateTime().atZone(ZoneId.systemDefault()).toInstant());
				SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
				e.setDateTime(sdf.format(time));
				utils.setScaleBySymbol(e, e.getSymbolName());
				ContractTradeWebsocketVO vo = new ContractTradeWebsocketVO();
				vo.setDateTime(e.getDateTime());
				vo.setSide(e.getSide());
				vo.setNum(e.getNum().toPlainString());
				vo.setPrice(e.getPrice().toPlainString());
				websocketVO.add(vo);
			} catch (Exception exception) {
				exception.printStackTrace();
			}
		});
		return JSONArray.parseArray(JSON.toJSONString(websocketVO));
	}

	private JSONObject getDepthMap(List<String> dataList) {
		String symbol = "";
		String level = "";
		symbol = dataList.get(1);
		level = dataList.get(3);
		Object depth_more = redis.opsForValue().get("depth:" + symbol + ":more");
		Object depth_empty = redis.opsForValue().get("depth:" + symbol + ":empty");
		JSONObject depth = new JSONObject();

		if (depth_more != null) {
			JSONObject jsonMore = JSONObject.parseObject(depth_more.toString());

			String depthString = jsonMore.get(level).toString();
			Gson gson = new Gson();
			TreeMap<BigDecimal, BigDecimal> m = gson.fromJson(depthString, new TypeToken<TreeMap<BigDecimal, BigDecimal>>() {
			}.getType());

			JSONArray jsonArray_more = new JSONArray();
			m.forEach((k, v) -> {
				JSONArray arr = new JSONArray();
				arr.add(k);
				arr.add(v);
				jsonArray_more.add(arr);
			});
			depth.put(ContractOrderEnum.Side.MORE.getValue(), jsonArray_more);
		}
		if (depth_empty != null) {
			JSONObject jsonEmpty = JSONObject.parseObject(depth_empty.toString());

			String depthString = jsonEmpty.get(level).toString();

			Gson gson = new Gson();
			TreeMap<BigDecimal, BigDecimal> m = gson.fromJson(depthString, new TypeToken<TreeMap<BigDecimal, BigDecimal>>() {
			}.getType());
			JSONArray jsonArray_empty = new JSONArray();
			m.forEach((k, v) -> {
				JSONArray arr = new JSONArray();
				arr.add(k);
				arr.add(v);
				jsonArray_empty.add(arr);
			});
			depth.put(ContractOrderEnum.Side.EMPTY.getValue(), jsonArray_empty);
		}
		return depth;
	}
}
